package com.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo3DSLApi {
  def main(args: Array[String]): Unit = {
    /**
     * 创建spark sql 环境
     *
     */

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("dsl")
      //指定spark sql  在shuffle之后分区数，默认时200， 类似hive中设计reduce的数量
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    /**
     * DSL Api
     */

    //读取一个json格式的文件
    val studentDF: DataFrame = spark.read
      .format("json") //指定读取数据的格式为json格式
      .load("data/students.json")

    studentDF.printSchema()

    /**
     * show: 查看df中数据, 相当于rdd中的action算子，会触发任务的执行
     *
     */

    studentDF.show()
    //指定打印多少行
    studentDF.show(100)
    //完整打印每一列的数据
    studentDF.show(false)

    /**
     * select : 选择数据，和sql中的select 用法基本一致，
     * dsl中select 不能使用聚合函数，需要在ogg中使用聚合函数
     * select相当于rdd中的转换算子
     *
     */

    //使用列名选择数据
    studentDF.select("id", "age") //.show()
    //使用sql 表达式
    studentDF.selectExpr("name", "age + 1 as age") //.show()

    //但如隐式转换，才能使用列对象
    import spark.implicits._
    //使用列对象的方式
    studentDF.select($"id", $"age") //.show()

    //使用列对象，可以直接进行sql计算
    studentDF.select($"id", $"age" + 1 as "age") //.show()

    //需要导入spark sql的函数，才能在DSl中使用函数
    import org.apache.spark.sql.functions._
    //在select中使用ql的函数
    studentDF.select($"id", substring($"clazz", 0, 2) as "cc") //.show()


    /**
     * where: 过滤数据
     */

    //字符串sql表达式
    studentDF.where("gender = '女' and age = 23") //.show()
    //使用列对象
    studentDF.where($"gender" =!= "男" and $"age" === 22) //.show()
    //使用函数
    studentDF.where(substring($"clazz", 0, 2) === "文科") //.show()


    /**
     * groupBy ogg ： 分组聚合, 分组和聚合要一起使用
     * 分组聚合之后返回的DF只包含分组字段和聚合字段的
     * 能使用的聚合函数和sql中是一样的
     *
     */

    studentDF
      .groupBy($"clazz")
      .agg(count($"clazz") as "num", round(avg($"age"), 2) as "avgAge")
    //.show()


    /**
     * order by : 排序
     *
     */

    studentDF
      .groupBy($"clazz")
      .agg(count($"clazz") as "num")
      .orderBy($"num".desc)
    //.show()


    /**
     * join: 表关联
     *
     */

    //读取分数表
    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,cid STRING, sco DOUBLE") //指定列名
      .load("data/score.txt")


    //关联字段名不一样的写法
    //val joinDF: DataFrame = studentDF.join(scoreDF, $"id" === $"sid", "inner")
    //关联字段名一样写法
    val joinDF: DataFrame = studentDF.join(scoreDF, "id")
    joinDF.show()


    /**
     * 开窗函数
     * 统计每个班级总分前十的学生
     *
     * withColumn: 在DF的基础上增加新的列
     *
     */
    joinDF
      .groupBy($"id", $"clazz") //按照学号和班级分组
      .agg(sum($"sco") as "sumSco") //计算总分
      //.select($"id", $"clazz", $"sumSco", row_number() over (Window.partitionBy($"clazz").orderBy($"sumSco".desc)))
      //简写，在前面的基础上增加列
      .withColumn("r", row_number() over Window.partitionBy($"clazz").orderBy($"sumSco".desc))
      //取前十
      .where($"r" <= 10)
      .show()

    /**
     * sql
     *
     */

    joinDF.createOrReplaceTempView("student_score")

    spark.sql(
      """
        |
        |select * from (
        |select id as sid,clazz,sumSco,row_number() over(partition by clazz order by sumSco desc) as r from (
        |select id,clazz,sum(sco) as  sumSco
        |from student_score
        |group by id,clazz
        |) as a
        |) as b
        |where r<=10
        |
        |""".stripMargin).show()
  }

}
